package cn.com.bluemoon.daps.sync.service.impl;

import cn.com.bluemoon.daps.common.domain.UserInfoHolder;
import cn.com.bluemoon.daps.common.enums.SyncStatus;
import cn.com.bluemoon.daps.common.enums.SyncToEnv;
import cn.com.bluemoon.daps.common.exception.DapThrowException;
import cn.com.bluemoon.daps.common.sql.layer.ILayerCheck;
import cn.com.bluemoon.daps.common.sql.layer.LayerCheckFactory;
import cn.com.bluemoon.daps.common.toolkit.BmAssetUtils;
import cn.com.bluemoon.daps.common.toolkit.SshUtils;
import cn.com.bluemoon.daps.model.entity.DapDataModel;
import cn.com.bluemoon.daps.sync.base.BmScript;
import cn.com.bluemoon.daps.sync.base.CompareResult;
import cn.com.bluemoon.daps.sync.base.ModifyType;
import cn.com.bluemoon.daps.sync.base.SqlCompareResult;
import cn.com.bluemoon.daps.sync.config.PtServerConfig;
import cn.com.bluemoon.daps.sync.core.sync.GlobalConfig;
import cn.com.bluemoon.daps.sync.core.sync.TableModel;
import cn.com.bluemoon.daps.sync.core.sync.compare.TableCompareRes;
import cn.com.bluemoon.daps.sync.core.sync.executor.SqlExecutor;
import cn.com.bluemoon.daps.sync.core.sync.executor.SqlExecutors;
import cn.com.bluemoon.daps.sync.core.sync.log.LogUtils;
import cn.com.bluemoon.daps.sync.core.sync.log.SyncEnvProdLog;
import cn.com.bluemoon.daps.sync.core.sync.log.SyncEnvProdLogEvent;
import cn.com.bluemoon.daps.sync.core.sync.sql.mysql.MysqlConnect;
import cn.com.bluemoon.daps.sync.core.sync.sql.mysql.MysqlCreateTableSql;
import cn.com.bluemoon.daps.sync.entity.DapSyncProdEnvTask;
import cn.com.bluemoon.daps.sync.entity.DapSyncProdEnvTaskDetail;
import cn.com.bluemoon.daps.sync.feignclient.DapModelModuleFacade;
import cn.com.bluemoon.daps.sync.service.*;
import cn.com.bluemoon.daps.sync.vo.model2prod.SyncModelOrderOut;
import cn.com.bluemoon.daps.sync.vo.model2prod.SyncModelOrderPageVo;
import cn.com.bluemoon.daps.sync.vo.model2prod.SyncOperRecordOut;
import cn.com.bluemoon.daps.sync.vo.model2prod.SyncPublishInfoOut;
import cn.com.bluemoon.daps.system.entity.DapSystemDatasourceEnvironment;
import cn.hutool.db.Db;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.metadata.IPage;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.context.event.EventListener;
import org.springframework.stereotype.Service;
import reactor.util.function.Tuple2;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.time.LocalDateTime;
import java.util.*;
import java.util.function.Function;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static cn.com.bluemoon.daps.sync.core.sync.log.SyncEnvProdLog.getTaskAndDetailId;

/**
 * 发布同步
 * 同步预发布
 * 同步生产
 * 流程：1、通过申请发布后流入同步任务
 * 2、同步过程：
 * 1）初始化同步基础数据{@link GlobalConfig}
 * 2）拉取待同步环境库表结构，到沙箱环境（测试与生产连接信息分别管理）
 * 3）沙箱模拟同步执行
 * 4）模拟通过，进入真实同步环境
 * 5）同步情况处理：发送邮件，记录日志，管理任务同步情况
 *
 * @author Jarod.Kong
 */
@Service
@Slf4j
public class DapModel2ProdSyncServiceImpl implements DapModel2ProdSyncService {
    @Resource
    private DapSyncProdEnvTaskService taskService;

    @Resource
    private DapSyncProdEnvTaskDetailService taskDetailService;

    @Resource
    private DapSyncBasicService syncBasicService;

    @Resource
    private DapSyncTaskOperRecordService operRecordService;

    @Resource
    private DapModelModuleFacade modelModuleFacade;

    @Resource
    private PtServerConfig ptServerConfig;

    private GlobalConfig.TableIdentity getTableIdentityByTaskDetail(List<DapSyncProdEnvTaskDetail> taskDetails) {
        // 2021/10/29 增加同步哪些表，用于指定导出目标环境到沙箱中的表
        GlobalConfig.TableIdentity tableIdentity = taskDetails.stream()
                .map((Function<DapSyncProdEnvTaskDetail, Map<String, List<String>>>) taskDetail -> {
                    return new HashMap<String, List<String>>(3) {{
                        put("tableIds", new ArrayList<String>() {{
                            add(taskDetail.getTableId());
                        }});
                        put("tableNames", new ArrayList<String>() {{
                            add(taskDetail.getTableName());
                        }});
                    }};
                }).reduce((v1, v2) -> {
                    v1.get("tableIds").addAll(v2.get("tableIds"));
                    v1.get("tableNames").addAll(v2.get("tableNames"));
                    return v1;
                })
                .map(v -> new GlobalConfig.TableIdentity(
                        v.get("tableIds").stream().distinct().collect(Collectors.toList()),
                        v.get("tableNames").stream().distinct().collect(Collectors.toList())))
                .orElse(null);
        return tableIdentity;
    }

    /**
     * 同步列表
     *
     * @param searchVo 查询条件
     * @return page
     */
    @Override
    public IPage<SyncModelOrderOut> querySyncList(SyncModelOrderPageVo<DapSyncProdEnvTask> searchVo) {
        String modelId = searchVo.getModelId();
        modelModuleFacade.getByIdAsset(modelId, "模型不存在");
        // 查询条件
        LambdaQueryWrapper<DapSyncProdEnvTask> where = taskService.bmLambdaQuery()
                .like(StringUtils.isNotBlank(searchVo.getChangeDescTrim()), DapSyncProdEnvTask::getChangeDesc, searchVo.getChangeDescTrim())
                .eq(DapSyncProdEnvTask::getModelId, modelId)
                .eq(searchVo.getPreProdStatus() != null && !SyncStatus.SYNC_ALL.equals(searchVo.getPreProdStatus()), DapSyncProdEnvTask::getPreProdStatus, searchVo.getPreProdStatus())
                .eq(searchVo.getProdStatus() != null && !SyncStatus.SYNC_ALL.equals(searchVo.getProdStatus()), DapSyncProdEnvTask::getProdStatus, searchVo.getProdStatus())
                .orderByDesc(DapSyncProdEnvTask::getChangeOrder)
                .orderByDesc(DapSyncProdEnvTask::getUpdateTime);
        IPage<DapSyncProdEnvTask> page = taskService.page(searchVo.getMpPage(), where);
        return page.convert(t -> SyncModelOrderOut.builder()
                .taskId(t.getId()).modelId(t.getModelId())
                .changeOrder(t.getChangeOrder()).changeDesc(t.getChangeDesc())
                .applicationBy(t.getApplicationBy())
                .syncBy(t.getSyncByOrNull()).preProdSyncStatus(t.getPreProdStatus().getCode()).prodSyncStatus(t.getProdStatus().getCode())
                .syncTime(t.getSyncTimeOrNull())
                .build());
    }

    /**
     * 发布同步
     *
     * @param syncToEnv 同步环境
     * @param modelId   模型id
     * @param taskId    模型单号任务id
     * @return 结果
     */
    @Override
    public boolean syncPublish(SyncToEnv syncToEnv, String modelId, String taskId, UserInfoHolder.UserInfo userInfo) {
        // 判断模型下是否存在该单号任务
        BmAssetUtils.isTrue((SyncToEnv.PROD.equals(syncToEnv) || SyncToEnv.PRE_PROD.equals(syncToEnv)), "目前支持同步预发布和生产环境");
        // 增加若预发布不是已同步，进行生产同步，则报错
        DapSyncProdEnvTask preTask = taskService.getByIdAsset(taskId, "同步任务不存在");
        if (syncToEnv.equals(SyncToEnv.PROD)) {
            BmAssetUtils.isTrue(SyncStatus.SYNC_DONE.equals(preTask.getPreProdStatus()), "请先完成预发布同步");
        }
        final String logBizKey = LogUtils.generateBizKey(modelId, taskId, syncToEnv);
        LogUtils.logAndSet(logBizKey, log, "进入发布同步-%s...", syncToEnv.getDescEnv());
        // 模型基础数据
        GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(modelId, syncToEnv, userInfo, true, null, logBizKey);
        LogUtils.logAndSet(logBizKey, log, "初始化配置完成.");
        globalConfig.setCheckSqlLayer(true);// 格式化
        boolean finalSyncStatus = false;
        List<TableCompareRes<SqlCompareResult>> res = getTableCompareRes(globalConfig, taskId);
        BmAssetUtils.isFalse(res.isEmpty(), "对比结果为空，不进行同步");
        LogUtils.logAndSet(logBizKey, log, "获取执行脚本完成.");
        try {
            // 校验+更新同步状态（若校验不同步不进行修改同步状态）
            this.updateTaskSyncDoing(syncToEnv, taskId, userInfo.getAccount(), globalConfig);
            LogUtils.logAndSet(logBizKey, log, "校验和更新同步状态完成.");

            boolean sandboxMock = false;
            operRecordService.mark(modelId, taskId, String.format("进入同步%s", syncToEnv.getDescEnv()));
            // 重建沙箱环境
            operRecordService.mark(modelId, taskId, String.format("同步%s-沙箱环境开始，执行人：%s", syncToEnv.getDescEnv(), userInfo.getUserName()));

            LogUtils.logAndSet(logBizKey, log, "开始重建沙箱环境...");
            rebuildSandboxEnv(globalConfig);
            LogUtils.logAndSet(logBizKey, log, "完成重建沙箱环境.");

            LogUtils.logAndSet(logBizKey, log, "开始启用沙箱环境，模拟%s脚本执行...", syncToEnv.getDescEnv());
            sandboxMock = doSyncOnSandbox(res, taskId, globalConfig);
            LogUtils.logAndSet(logBizKey, log, "启用沙箱环境，模拟%s脚本执行，执行结果：%s.", syncToEnv.getDescEnv(), (sandboxMock ? "成功" : "失败"));

            operRecordService.mark(modelId, taskId, String.format("同步%s-沙箱环境结束，同步结果：%s", syncToEnv.getDescEnv(), (sandboxMock ? "成功" : "失败")));
            // 必须沙箱成功才可以真正同步
            if (!sandboxMock) {
                // 沙箱报错情况下，把单号下表脚本已同步情况的都更新为未同步（用户不知道有沙箱环境的执行） 去掉，原因：强行更改沙箱执行表的状态会导致同步管理的表环境状态为未同步，导致无法更改状态。
                taskDetailService.updateSyncStatusByTaskIdAndSyncEnvAndSyncStatus(taskId, syncToEnv, SyncStatus.SYNC_ERROR);
                taskService.updateSyncFailOnSandbox(taskId, true);
                taskDetailService.updateTableSyncFailOnSandbox(taskId, true);
                log.error("同步{}中沙箱环境执行失败，更新taskId={}数据", globalConfig.getSyncToEnv().getDesc(), taskId);
            } else {
                operRecordService.mark(modelId, taskId, String.format("同步%s开始，执行人：%s", syncToEnv.getDescEnv(), userInfo.getUserName()));
                LogUtils.logAndSet(logBizKey, log, "同步%s开始，执行人：%s", syncToEnv.getDescEnv(), userInfo.getUserName());
                finalSyncStatus = doSyncOnPublish(res, taskId, globalConfig);
//                operRecordService.mark(modelId, taskId, String.format("同步%s结束，同步结果：%s", syncToEnv.getDescEnv(), (finalSyncStatus ? "成功" : "失败")));
                // 清除日志头内容
                LogUtils.clearByKey(logBizKey);
                LogUtils.logAndSet(logBizKey, log, "同步%s结束，同步结果：%s", syncToEnv.getDescEnv(), (finalSyncStatus ? "成功" : "失败"));
            }
        } catch (Exception e) {
            finalSyncStatus = false;
            log.error("发布同步异常，", e);
            LogUtils.logAndSet(logBizKey, log, "同步%s异常，详情：%s", syncToEnv.getDescEnv(), ExceptionUtils.getMessage(e));
        } finally {
            // 重新查出来
            DapSyncProdEnvTask task = taskService.getById(taskId);
            // 修改单号状态为 同步失败；而单号ddl数据由cn.com.bluemoon.daps.sync.service.impl.DapModel2ProdSyncServiceImpl.syncEnvLogEventHandler维护
            task.setSyncStatus(finalSyncStatus ? SyncStatus.SYNC_DONE : SyncStatus.SYNC_ERROR, globalConfig.getSyncToEnv());
            task.setSyncTime(LocalDateTime.now(), globalConfig.getSyncToEnv());
            boolean status = task.updateById();
            operRecordService.mark(modelId, taskId, String.format("同步%s结束，同步结果：%s", syncToEnv.getDescEnv(), (finalSyncStatus ? "成功" : "失败")));
            log.info("同步{}执行情况：status:{}，更新单号{}数据为{}", globalConfig.getSyncToEnv().getDesc(), finalSyncStatus, taskId, status);
            log.error("同步{}执行情况：status:{}，发送错误通知", globalConfig.getSyncToEnv().getDesc(), finalSyncStatus);
            syncBasicService.sendEmailPublishSync(task, globalConfig);
            LogUtils.logAndSet(logBizKey, log, "同步%s执行完成，已发送通知", syncToEnv.getDescEnv());
            log.info("最终同步到{}环境情况：{}", syncToEnv.getDesc(), finalSyncStatus);
            String logContent = LogUtils.getLogByKey(logBizKey);
            log.info("log:{}", logContent);
            taskDetailService.updateSyncLogContent(taskId, syncToEnv, logContent);
            LogUtils.clearByKey(logBizKey);
        }
        return finalSyncStatus;
    }

    /**
     * 模型+任务id（获取单号）
     * 获取操作记录
     *
     * @param modelId 模型
     * @param taskId  任务id
     * @return 记录
     */
    @Override
    public List<SyncOperRecordOut> findOperRecords(String modelId, String taskId) {
        DapSyncProdEnvTask task = taskService.getByIdAsset(taskId, "获取模型任务失败");
        BmAssetUtils.isTrue(task.getModelId().equals(modelId), "模型下无法找到该任务");
        return operRecordService.findOpersByTaskId(modelId, taskId);
    }

    /**
     * 模型+任务id（获取单号）
     * 获取发布明细
     *
     * @param modelId   模型
     * @param taskId    任务id
     * @param syncToEnv 查看哪个环境数据
     * @return 记录
     */
    @Override
    public SyncPublishInfoOut getPublishInfo(String modelId, String taskId, SyncToEnv syncToEnv) {
        GlobalConfig globalConfig = syncBasicService.buildSyncGlobalConfig(modelId, syncToEnv);
        DapSyncProdEnvTask task = taskService.getByIdAsset(taskId, "获取模型任务失败");
        BmAssetUtils.isTrue(task.getModelId().equals(modelId), "模型下无法找到该任务");
        BmAssetUtils.isTrue(SyncToEnv.PRE_PROD.equals(syncToEnv) || SyncToEnv.PROD.equals(syncToEnv), "只允许获取模型预发布或生产环境任务信息");
        DapDataModel model = globalConfig.getModel();
        DapSystemDatasourceEnvironment dataSourceEnv = globalConfig.getDataSourceEnv();
        List<DapSyncProdEnvTaskDetail> details = taskDetailService.findTaskDetailsByTaskId(taskId).stream().peek(d -> d.setDdl(d.getDdl().replace(BmScript.SQL_SPLIT_CHAR_RL, ";"))).collect(Collectors.toList());
        List<DapSyncProdEnvTaskDetail> doneDdls = details.stream().filter(d -> SyncStatus.SYNC_DONE.equals(d.getScriptSyncStatus(syncToEnv))).collect(Collectors.toList());
        List<String> ddls = doneDdls.stream().map(DapSyncProdEnvTaskDetail::getDdl).collect(Collectors.toList());
        List<String> unDdls = details.stream().filter(d -> !SyncStatus.SYNC_DONE.equals(d.getScriptSyncStatus(syncToEnv))).map(DapSyncProdEnvTaskDetail::getDdl)
                .collect(Collectors.toList());
        // 同步执行失败+在沙箱环境执行失败才调整未执行+已执行内容
        if (SyncStatus.SYNC_ERROR.equals(task.getSyncStatus(syncToEnv))
                && Boolean.TRUE.equals(task.getSandboxSyncFail())) {
            List<String> ddlsFix = new ArrayList<>();
            for (DapSyncProdEnvTaskDetail d : doneDdls) {
                if (Boolean.TRUE.equals(d.getSandboxSyncFail())) {
                    unDdls.add(d.getDdl());
                } else {
                    ddlsFix.add(d.getDdl());
                }
            }
            ddls.clear();
            ddls.addAll(ddlsFix);
            log.warn("当前执行环境在沙箱中失败，把已执行的脚本都归到未执行中显示，同时情况已执行状态集合数据");
        }
        SyncPublishInfoOut publishInfoOut = SyncPublishInfoOut.builder()
                .modelName(model.getName()).dbname(dataSourceEnv.getDatabaseName())
                .applicationBy(task.getApplicationBy()).applicationTime(task.getApplicationTime()).verifyBy(task.getVerifyBy())
                .changeDesc(task.getChangeDesc()).changeOrder(task.getChangeOrder())
                .taskId(taskId).host(dataSourceEnv.getHost()).port(dataSourceEnv.getPort())
                .syncBy(task.getSyncBy(syncToEnv)).syncTime(task.getSyncTime(syncToEnv))
                .syncStatus(task.getSyncStatus(syncToEnv).getCode())
                .syncDdls(ddls).unSyncDdls(unDdls)
                .canSync(SyncStatus.SYNC_WAIT.equals(task.getSyncStatus(syncToEnv)) || SyncStatus.SYNC_ERROR.equals(task.getSyncStatus(syncToEnv)))
                .build();
        log.info("获取模型={}，任务单号：{}，的同步信息", model, task.getChangeOrder());
        return publishInfoOut;
    }

    /**
     * 模型+任务id（获取单号）
     * 任务明细、对应环境的、日志汇总
     * 1.已同步的
     * 2.同步中的
     * 3.未同步的（在沙箱环境中执行，不调整ddl任务状态，所以为未同步）
     * 总，把不为空的日志进行获取，然后整合
     *
     * @param syncToEnv 查看哪个环境数据
     * @param modelId   模型
     * @param taskId    任务id
     * @return 记录
     */
    @Override
    public List<String> getTaskLogs(SyncToEnv syncToEnv, String modelId, String taskId) {
        DapSyncProdEnvTask task = taskService.getByIdAsset(taskId, "获取模型任务失败");
        BmAssetUtils.isTrue(task.getModelId().equals(modelId), "模型下无法找到该任务");
        BmAssetUtils.isTrue(SyncToEnv.PRE_PROD.equals(syncToEnv) || SyncToEnv.PROD.equals(syncToEnv), "只允许获取模型预发布或生产环境任务信息");
        // 增加如果为未同步则返回空
        if (SyncStatus.SYNC_WAIT.equals(task.getSyncStatus(syncToEnv))) {
            log.info("taskId={}，未同步，无日志数据", taskId);
            return Collections.emptyList();
        }
        List<DapSyncProdEnvTaskDetail> details = taskDetailService.findTaskDetailsByTaskId(taskId);
        return details.stream().map(t -> {
            // 若syncLog 存在则不ssh查看日志文件
            String syncLog = t.getSyncLog(syncToEnv);
            String syncLogPath = t.getSyncLogPath();
            if (StringUtils.isBlank(syncLog) && SyncStatus.SYNC_DOING.equals(t.getScriptSyncStatus(syncToEnv))) {
                Tuple2<Boolean, String> tuple2 = SshUtils.client().sshInfo(ptServerConfig.getPtSsh()).runScript("cat " + syncLogPath);
                if (Boolean.TRUE.equals(tuple2.getT1())) {
                    syncLog = tuple2.getT2();
                } else {
                    log.error("查看日志文件{}出错！", syncLogPath);
                }
            }
            if (StringUtils.isNotBlank(syncLog)) {
                return String.format("\n>>>>同步脚本%s表ddl执行情况，如下：\n %s", t.getTableName(), syncLog);
            }
            return null;
        }).filter(StringUtils::isNotBlank).collect(Collectors.toList());
    }

    /**
     * 更新任务单号的同步状态为同步中&更新同步时间 & 同步人
     *
     * @param syncToEnv 环境
     * @param taskId    任务id
     * @param account   同步人
     */
    private void updateTaskSyncDoing(SyncToEnv syncToEnv, String taskId, String account, GlobalConfig globalConfig) {
        // 模型任务
        DapSyncProdEnvTask task = taskService.getByIdAsset(taskId, "模型单号任务不存在");
        this.checkSyncTaskValid(syncToEnv, task, globalConfig);
        // 执行前修改为同步中
        task.setSyncStatus(SyncStatus.SYNC_DOING, syncToEnv);
        task.setSyncTime(LocalDateTime.now(), syncToEnv);
        task.setSyncBy(account, syncToEnv);
        boolean updateStatus = task.updateById();
        log.info("修改任务单号状态为{}, status:{}", SyncStatus.SYNC_DOING, updateStatus);
    }

    /**
     * 同步前检查 1）是否只有当前模型单号为未同步 2）模型是否找到该模型单号 3）所选任务是否为未同步
     *
     * @param syncToEnv 环境
     * @param task      任务
     */
    private void checkSyncTaskValid(SyncToEnv syncToEnv, DapSyncProdEnvTask task, GlobalConfig globalConfig) {
        String taskId = task.getId();
        String modelId = task.getModelId();
        BmAssetUtils.isTrue(task.getModelId().equals(modelId), "模型无法找到该模型单号");
        // 判断模型下存在其他单号任务出于非已同步状态，则不允许执行
        LambdaQueryWrapper<DapSyncProdEnvTask> neDoneWrapper = taskService.bmLambdaQuery()
                .eq(DapSyncProdEnvTask::getModelId, modelId)
                .ne(DapSyncProdEnvTask::getId, taskId)
                .ne(DapSyncProdEnvTask::getPreProdStatus, SyncStatus.SYNC_DONE)
                .ne(DapSyncProdEnvTask::getProdStatus, SyncStatus.SYNC_DONE);
        List<DapSyncProdEnvTask> notDoneTasks = taskService.list(neDoneWrapper);
        BmAssetUtils.isTrue(notDoneTasks.isEmpty(), "当前模型发布队列存在未同步生产库数据，请先执行同步");
        if (SyncToEnv.PRE_PROD.equals(syncToEnv)) {
            BmAssetUtils.isTrue(SyncStatus.SYNC_WAIT.equals(task.getPreProdStatus())
                    || SyncStatus.SYNC_ERROR.equals(task.getPreProdStatus()), "所选任务非未同步或同步失败");
        }
        if (SyncToEnv.PROD.equals(syncToEnv)) {
            BmAssetUtils.isTrue(SyncStatus.SYNC_WAIT.equals(task.getProdStatus())
                    || SyncStatus.SYNC_ERROR.equals(task.getProdStatus()), "所选任务非未同步");
            // 2021/7/9 2021/7/12 增加同步生产环境的前提条件是：在模型的预发布环境存在的情况下，同步预发布成功
            List<DapSystemDatasourceEnvironment> allEnv = globalConfig.getDataSourceAndEnv().getAllEnv();
            if (allEnv.stream().anyMatch(d -> d.getEnvironment() != null && d.getEnvironment().equals(3))) {
                BmAssetUtils.isTrue(SyncStatus.SYNC_DONE.equals(task.getPreProdStatus()), "请先同步预发布环境");
            }
        }
    }

    /**
     * 对沙箱环境进行删除+创建数据库
     */
    private void dropAndCreateSandboxDatabase(GlobalConfig globalConfig) throws DapThrowException {
        try {
            MysqlConnect env = globalConfig.getSandboxMysqlConnect();
            final String url = String.format("jdbc:%s://%s:%d/%s?createDatabaseIfNotExist=true", env.getDbTypeName(), env.getHost(),
                    env.getPort(), ""),
                    username = env.getUser(),
                    password = env.getPassword();
            String dbname = env.getDatabase();
            BmAssetUtils.notNull(dbname, "数据库名不可为空");
            try (Connection conn = DriverManager.getConnection(url, username, password)) {
                Arrays.asList("drop database if exists " + dbname, "create database if not exists " + dbname)
                        .forEach(sql -> {
                            try (PreparedStatement ps = conn.prepareStatement(sql)) {
                                boolean execute = ps.execute();
                                log.warn("重建库-执行{}，情况：{}", sql, execute);
                            } catch (SQLException e) {
                                e.printStackTrace();
                            }
                        });
            }
        } catch (SQLException e) {
            throw new DapThrowException("重建沙箱环境失败", e);
        }
    }

    /**
     * 导入同步环境的建表语句 & 导入到沙箱中提供mock
     *
     * @param globalConfig 全局配置
     */
    private void dumpFileAndImportFileToSandboxEnv(GlobalConfig globalConfig) throws DapThrowException {
        // 同步沙箱导入到的数据库前，删除历史的表和数据库名
        SyncToEnv syncToEnv = globalConfig.getSyncToEnv();
        DapSystemDatasourceEnvironment ds = globalConfig.getDataSourceEnv();
        // 2021-07-08 导出前尝试创建环境数据库明
        boolean initDbName = globalConfig.initDbName(false);
        log.warn("拉取同步环境数据中尝试初始化数据库名，status:{}，若已存在或者创建失败为false", initDbName);
        // 2021/10/29 增加导出指定表
        List<String> tables = Optional.ofNullable(globalConfig.getTableIdentity()).map(GlobalConfig.TableIdentity::getTableNames).orElse(null);
        if (tables == null || tables.isEmpty()) {
            log.warn("重建库中发现没有指定导入相关表，默认会导出整个库所有表, 数据源id={}", ds.getDatasourceId());
        }
        List<TableModel> syncEnvTables = syncBasicService.getMetadataTableModel(globalConfig);
        if (syncEnvTables.isEmpty()) {
            log.warn("拉取目标环境{}，表为空", syncToEnv);
            return;
        }
        try {
            List<String> createTableSqls = syncEnvTables.stream().flatMap(t -> t.getCreateTableSqls().stream()).collect(Collectors.toList());
            Db sandboxDb = globalConfig.findSandboxDb();
            createTableSqls.parallelStream().forEach(createTableSql -> {
                try {
                    int execute = sandboxDb.execute(createTableSql);
                    log.debug("拉取目标环境,{},表{}导入沙箱情况：{}", syncToEnv, createTableSql, execute);
                } catch (SQLException e) {
                    log.error("{}导入目标环境到沙箱环境中失败", createTableSql);
                }
            });
        } catch (SQLException e) {
            throw new DapThrowException("导入目标环境到沙箱环境中失败,", e);
        }
    }

    /**
     * 进行发布同步-正式环境
     *
     * @param taskId       任务id
     * @param globalConfig 全局配置
     * @return 同步结果
     */
    private boolean doSyncOnPublish(List<TableCompareRes<SqlCompareResult>> res, String taskId, GlobalConfig globalConfig) {
        globalConfig.setSyncSandboxMock(false);
        return doSync0(res, taskId, globalConfig);
    }

    /**
     * 进行发布同步-沙箱环境
     *
     * @param res          结果集
     * @param taskId       任务id
     * @param globalConfig 全局配置
     * @return 同步结果
     */
    private boolean doSyncOnSandbox(List<TableCompareRes<SqlCompareResult>> compareRes, String taskId, GlobalConfig globalConfig) {
        globalConfig.setSyncSandboxMock(true);
        return doSync0(compareRes, taskId, globalConfig);
    }

    /**
     * 重建沙箱环境-先删除再重建数据库 & 先到导出正式环境建表语句再导入到沙箱中
     *
     * @param globalConfig 全局配置
     */
    private void rebuildSandboxEnv(GlobalConfig globalConfig) throws DapThrowException {
        dropAndCreateSandboxDatabase(globalConfig);
        dumpFileAndImportFileToSandboxEnv(globalConfig);
    }

    /**
     * 进行沙箱、预发布、生产的同步核心入库
     *
     * @param taskId       任务id 获取任务明细数据
     * @param globalConfig 全局配置
     * @return 结果
     */
    private boolean doSync0(List<TableCompareRes<SqlCompareResult>> res, String taskId, GlobalConfig globalConfig) {
        // 根据模型对应的数据源中的开关（是否开启ptosc同步）
        Boolean isPtosc = globalConfig.getDataSource().getIsPtosc();
        SqlExecutor.ExecutorType type;
        if (isPtosc == null || Boolean.TRUE.equals(isPtosc)) {
            type = SqlExecutor.ExecutorType.PT;
        } else {
            type = SqlExecutor.ExecutorType.JDBC;
        }
        LogUtils.logAndSet(LogUtils.generateBizKey(globalConfig.getModel().getId(), taskId, globalConfig.getSyncToEnv()), log, "同步%s，选用执行方式：%s", globalConfig.getSyncToEnv().getDescEnv(), type);
        return SqlExecutors.use(type, globalConfig).sqlResult0(res).execute();
    }

    /**
     * 获取任务下的ddl明细数据
     *
     * @param taskId       任务id 获取任务明细数据
     * @param globalConfig 全局配置
     * @return 获取任务下的ddl明细数据
     */
    private List<TableCompareRes<SqlCompareResult>> getTableCompareRes(GlobalConfig globalConfig, String taskId) {
        List<DapSyncProdEnvTaskDetail> taskDetails = taskDetailService.findSyncTasks(taskId, globalConfig);
        GlobalConfig.TableIdentity tableIdentity = getTableIdentityByTaskDetail(taskDetails);
        globalConfig.setTableIdentity(tableIdentity);
        Map<String, String> tableAndCreateSql = tableAndCreateSqlMap(globalConfig);
        return taskDetails.stream()
                .filter(t -> t.getPtDdl() != null || t.getPtOsc() != null) // 要是没脚本则忽略
                .flatMap((Function<DapSyncProdEnvTaskDetail, Stream<TableCompareRes<SqlCompareResult>>>) d -> {
                    final String createTableSql = tableAndCreateSql.getOrDefault(d.getTableName(), null);
                    List<TableCompareRes<SqlCompareResult>> tableCompareRes = new ArrayList<>();
                    final String taskAndDetailId = getTaskAndDetailId(d);
                    // 要是data存在ptddl则执行ptddl，后执行ptOsc => 调整为ptddl生成独立的CompareResult、ptosc生成独立的CompareResult
                    List<CompareResult<SqlCompareResult>> compareResults = new ArrayList<>(2);
                    if (d.getPtDdl() != null) {
                        BmScript bmScript = BmScript.builder().sql(d.getPtDdl()).ptSql(null).fullSql(d.getDdl()).build();
                        SqlCompareResult c = SqlCompareResult.builder().script(bmScript).build();
                        CompareResult<SqlCompareResult> result = CompareResult.<SqlCompareResult>builder()
                                .modelId(d.getModelId())
                                .tableName(d.getTableName())
                                .tableId(d.getTableId())
                                .modifyType(null)// 无法确定，给null不影响后续ptosc的脚本执行
                                .modifyLevel(ModifyType.ModifyLevel.TABLE)// 2021年6月28日 区分表级脚本还是非表级
                                .data(c)
                                .build();
                        compareResults.add(result);
                    }
                    if (d.getPtOsc() != null) {
                        BmScript bmScript = BmScript.builder().sql(null).ptSql(d.getPtOsc()).fullSql(d.getDdl()).build();
                        SqlCompareResult c = SqlCompareResult.builder().script(bmScript).build();
                        CompareResult<SqlCompareResult> result = CompareResult.<SqlCompareResult>builder()
                                .modelId(d.getModelId())
                                .tableName(d.getTableName())
                                .tableId(d.getTableId())
                                .data(c)
                                .build();
                        compareResults.add(result);
                    }
                    CompareResult<SqlCompareResult> tableInfo = compareResults.get(0);
                    tableInfo.setTableCreateSql(createTableSql);
                    TableCompareRes<SqlCompareResult> t1 = TableCompareRes.<SqlCompareResult>builder()
                            .globalConfig(globalConfig)
                            .compareResults(compareResults)
                            .tableInfo(tableInfo)//前置条件必定至少一个
                            .uniqueKey(taskAndDetailId)
                            .build();
                    tableCompareRes.add(t1);
                    return tableCompareRes.stream();
                }).collect(Collectors.toList());
    }

    /**
     * 非沙箱环境获取元数据建表语句
     *
     * @param globalConfig 配置
     * @return sql
     */
    private Map<String, String> tableAndCreateSqlMap(GlobalConfig globalConfig) {
        // 增加同步环境中的当前建表语句
        List<TableModel> tableModels = syncBasicService.getMetadataTableModel(globalConfig);
        ILayerCheck layerCheck = LayerCheckFactory.type();
        Map<String, String> tableAndCreateSql = tableModels.stream().collect(Collectors.toMap(t -> t.getTable().getEngName(), a -> {
            MysqlCreateTableSql createTableSql = new MysqlCreateTableSql();
            createTableSql.setFields(a.getFields());
            createTableSql.setIndexes(a.getIndexes());
            createTableSql.setTable(a.getTable());
            return globalConfig.isCheckSqlLayer() ? layerCheck.getFormatMysqlSql(createTableSql.toString()) : createTableSql.toString();
        }, (a, b) -> b));
        return tableAndCreateSql;
    }

    /**
     * 调整单号状态
     * 调整单号明细数据状态
     *
     * @param envLogEvent 接收沙箱、预发布、生产的广播消息
     */
    @EventListener(classes = SyncEnvProdLogEvent.class)
    public void syncEnvLogEventHandler(SyncEnvProdLogEvent envLogEvent) {
        SyncEnvProdLog syncEnvLog = envLogEvent.getSource();
        final GlobalConfig globalConfig = envLogEvent.getGlobalConfig();
        log.info("处理消息，TYPE={}，模型：{}，表id：{}", envLogEvent.getType(), syncEnvLog.getModelId(), syncEnvLog.getTableId());
        String taskId = syncEnvLog.getTaskId();
        String taskDetailId = syncEnvLog.getTaskDetailId();
        if (taskId == null || taskDetailId == null) {
            log.warn("发布同步中，任务id和任务表id为空，不进行更新同步信息");
            return;
        }
        // 2021/7/1 如果 if (syncEnvLog.isSyncSandboxMock()) {}只修改执行路径+执行日志，不修改状态（出于前端需要查看日志，但是不修改ddl状态）
        final boolean syncSandboxMock = syncEnvLog.isSyncSandboxMock();
        SyncToEnv syncEnv = syncEnvLog.getSyncEnv();
        DapSyncProdEnvTask task = taskService.getByIdAsset(taskId, "同步任务不存在");
        DapSyncProdEnvTaskDetail taskDetail = taskDetailService.getByIdAsset(taskDetailId, "同步任务表日志不存在");
        taskDetail.setSyncScript(syncEnvLog.getSyncScriptStr());
        taskDetail.setSyncLogPath(syncEnvLog.getSyncLogPath());
        // 日志执行追加
        String preSyncLog = taskDetail.getSyncLog(syncEnv);
        String currSyncLog = syncEnvLog.getSyncLogStr(null);
        final String appendSyncLog;
        if (StringUtils.isBlank(preSyncLog)) {
            String preLogContent = LogUtils.getLogByKey(globalConfig.getLogBizKey());
            appendSyncLog = preLogContent + "\n" + currSyncLog;
        } else {
            appendSyncLog = preSyncLog + "\n" + currSyncLog;
        }
        taskDetail.setSyncLog(appendSyncLog, syncEnv);
        taskDetail.setScriptSyncStatus(syncEnvLog.getSyncStatus(), syncEnv);
        taskDetail.setNeedDba(SyncStatus.SYNC_ERROR.equals(syncEnvLog.getSyncStatus()));
        taskDetail.setSyncPreTbSql(syncEnvLog.getTableCreateSql(), syncEnv);
        boolean status = taskDetailService.saveOrUpdate(taskDetail);
        log.info("更新任务表同步信息：{}，status:{}", taskDetailId, status);
        // 修改模型单号
        if (syncSandboxMock) {
            log.warn("沙箱模拟执行不进行更新单号状态变更，由沙箱同步完成后返回的结果进行变更");
            return;
        }
        if (SyncToEnv.PRE_PROD.equals(syncEnv)) {
            // 最后出现taskId下的所有任务是否都执行完
            List<DapSyncProdEnvTaskDetail> list = taskDetailService.findTaskDetailsByTaskId(taskId);
            // 全部都为已同步为已同步，存在一个同步失败为同步失败，存在一个同步中为同步中，全部为未同步则未同步
            SyncStatus syncStatus = SyncStatus.SYNC_WAIT;
            if (list.stream().anyMatch(d -> SyncStatus.SYNC_DOING.equals(d.getPreProdScriptStatus()))) {
                syncStatus = SyncStatus.SYNC_DOING;
            }
            if (list.stream().allMatch(d -> SyncStatus.SYNC_DONE.equals(d.getPreProdScriptStatus()))) {
                syncStatus = SyncStatus.SYNC_DONE;
            }
            if (list.stream().anyMatch(d -> SyncStatus.SYNC_ERROR.equals(d.getPreProdScriptStatus()))) {
                syncStatus = SyncStatus.SYNC_ERROR;
            }
            task.setPreProdStatus(syncStatus);
            task.setPreProdSyncTime(syncEnvLog.getSyncTime());
            task.setPreProdSyncBy(syncEnvLog.getSyncExecBy());
            status = taskService.updateById(task);
            log.info("更新任务同步主表信息：{}，syncStatus : {}, status:{}", task.getId(), syncStatus, status);
        }
        if (SyncToEnv.PROD.equals(syncEnv)) {
            // 最后出现taskId下的所有任务是否都执行完
            List<DapSyncProdEnvTaskDetail> list = taskDetailService.findTaskDetailsByTaskId(taskId);
            // 全部都为已同步为已同步，存在一个同步失败为同步失败，存在一个同步中为同步中，全部为未同步则未同步
            SyncStatus syncStatus = SyncStatus.SYNC_WAIT;
            if (list.stream().anyMatch(d -> SyncStatus.SYNC_DOING.equals(d.getProdScriptStatus()))) {
                syncStatus = SyncStatus.SYNC_DOING;
            }
            if (list.stream().allMatch(d -> SyncStatus.SYNC_DONE.equals(d.getProdScriptStatus()))) {
                syncStatus = SyncStatus.SYNC_DONE;
            }
            if (list.stream().anyMatch(d -> SyncStatus.SYNC_ERROR.equals(d.getProdScriptStatus()))) {
                syncStatus = SyncStatus.SYNC_ERROR;
            }
            task.setProdStatus(syncStatus);
            task.setProdSyncTime(syncEnvLog.getSyncTime());
            task.setProdSyncBy(syncEnvLog.getSyncExecBy());
            status = taskService.updateById(task);
            log.info("更新任务同步主表信息：{}，syncStatus : {}, status:{}", task.getId(), syncStatus, status);
        }
    }


}
